package rxgo

import (
	"context"
	"reflect"
	"fmt"
	"time"
)

// 结构体函数 opFunc
type filterOperator struct {
	opFunc func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool)
}

// interface 的 reflect 类型转换
// 当 flip 非空，使用缓冲 buffer 进行数据传输
func (ftop filterOperator) op(ctx context.Context, o *Observable) {
	in := o.pred.outflow
	out := o.outflow
	go func() {
		end := false
		for x := range in {
			if end {
				break
			}
			// can not pass a interface as parameter (pointer) to gorountion for it may change its value outside!
			xv := reflect.ValueOf(x)
			// send an error to stream if the flip not accept error
			if e, ok := x.(error); ok && !o.flip_accept_error {
				o.sendToFlow(ctx, e, out)
				continue
			}
			if ftop.opFunc(ctx, o, xv, out) {
				end = true
			}
		}
		if o.flip != nil {
			// 缓存用于实现 Last, SkipLast, TakeLast 操作
			buffer := (reflect.ValueOf(o.flip))
			if buffer.Kind() != reflect.Slice {
				panic("flip is not a buffer")
			}
			for i := 0; i < buffer.Len(); i++ {
				o.sendToFlow(ctx, buffer.Index(i).Interface(), out)
			}
		}
		o.closeFlow(out)
	}()
}

// 模仿 newTransformObservable 写的初始化函数
func (parent *Observable) newFilterObservable(name string) (o *Observable) {
	//new Observable
	o = newObservable()
	o.Name = name

	//chain Observables
	parent.next = o
	o.pred = parent
	o.root = parent.root

	//set options
	o.buf_len = BufferLen
	return o
}

var count = 0

// Debounce：会过滤掉发射速率过快的数据项，实现数据的延迟发送
func (parent *Observable) Debounce(timespan time.Duration) (o *Observable) {
	o = parent.newFilterObservable("debounce")
	count = 0
	o.timeSpan = timespan
	o.operator = debounceOperator
	return o
}

// 只有当 Observable 在指定的时间后还没有发射数据时，才发射一个数据
var debounceOperator = filterOperator{
	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
		// 如果 count 改变了，证明有新数据
		count++
		go func() {
			tempCount := count
			time.Sleep(o.timeSpan)
			select {
			case <-ctx.Done():
				return
			default:
				if tempCount == count {
					o.sendToFlow(ctx, item.Interface(), out)
				}
			}
		}()
		return false
	},
}

// Distinct：抑制（过滤掉）重复的数据项，只允许还没有发射过的数据项通过
func (parent *Observable) Distinct() (o *Observable) {
	o = parent.newFilterObservable("distinct")
	o.disMap = make(map[string]bool)
	o.operator = distinctOperator
	return o
}

// 使用 map 来判断是否重复
var distinctOperator = filterOperator{
	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
		// 将 item 转化为相应的 string
		itemStr := fmt.Sprintf("%v", item)
		if _, ok := o.disMap[itemStr];  !ok {
			o.disMap[itemStr] = true
			o.sendToFlow(ctx, item.Interface(), out)
		}
		return false
	},
}

//ElementAt： 只发射第N项数据
func (parent *Observable) ElementAt(num int) (o *Observable) {
	o = parent.newFilterObservable("elementAt")
	count = 0
	o.elementAtNum = num
	o.operator = elementAtOperator
	return o
}

// 获取原始 Observable 发射的数据序列指定索引位置的数据项，然后当做自己的唯一数据发射
var elementAtOperator = filterOperator{
	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
		// 第n项数据
		if count == o.elementAtNum {
			o.sendToFlow(ctx, item.Interface(), out)
			return true
		}
		count++
		return false
	},
}

// First： 发射第一个数据
func (parent *Observable) First() (o *Observable) {
	o = parent.newFilterObservable("first")
	o.operator = firstOperator
	return o
}

// 传输完第一个数据后返回 true
var firstOperator = filterOperator{
	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
		o.sendToFlow(ctx, item.Interface(), out)
		return true
	},
}

// IgnoreElements： 不发射任何数据，只发射 Observable 的终止通知
func (parent *Observable) IgnoreElements() (o *Observable) {
	o = parent.newFilterObservable("ignoreElements")
	o.operator = ignoreOperator
	return o
}

// 什么都不做直接返回
var ignoreOperator = filterOperator{
	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
		return false
	},
}

// Last： 发射最后一个数据
func (parent *Observable) Last() (o *Observable) {
	o = parent.newFilterObservable("last")
	o.operator = lastOperator
	return o
}

var lastOperator = filterOperator{
	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
		o.flip = append([]interface{}{}, item.Interface())
		return false
	},
}

var latest interface{}

// Sample： 输出规定时间段中，最后到达的数据
func (parent *Observable) Sample(sample chan interface{}) (o *Observable) {
	o = parent.newFilterObservable("sample")
	latest = nil
	o.sample = sample
	o.operator = sampleOperator
	return o
}

// 定时查看一个 Observable，然后发射自上次采样以来它最近发射的数据
var sampleOperator = filterOperator{
	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
		latest = item.Interface()
		go func() {
			tempEnd := true
			for tempEnd {
				select {
				case <-ctx.Done():
					tempEnd = true

				case <-o.sample:
					if latest != nil {
						if o.sendToFlow(ctx, latest, out) {
							tempEnd = false
						}
						latest = nil
					}
				}
			}
		}()
		return false
	},
}

// Skip：跳过前几个数据
func (parent *Observable) Skip(num int) (o *Observable) {
	o = parent.newFilterObservable("skip")
	count = 0
	o.skipNum = num
	o.operator = skipOperator
	return o
}

// 忽略 Observable 发射的前N项数据，只保留之后的数据
var skipOperator = filterOperator{
	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
		count++
		if count > o.skipNum {
			o.sendToFlow(ctx, item.Interface(), out)
		}
		return false
	},
}

var lasts []interface{}

//SkipLast ： 跳过最后几个数据
func (parent *Observable) SkipLast(num int) (o *Observable) {
	o = parent.newFilterObservable("skipLast")
	count = 0
	o.skipLastNum = num
	lasts = nil
	o.operator = skipLastOperator
	return o
}

// 忽略 Observable 发射的后N项数据，只保留前面的数据。
var skipLastOperator = filterOperator{
	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
		if count == o.skipLastNum {
			o.sendToFlow(ctx, lasts[0], out)
			lasts = lasts[1:]
		} else {
			count++
		}
		lasts = append(lasts, item.Interface())
		return false
	},
}

//Take ：选取前几个元素输出
func (parent *Observable) Take(num int) (o *Observable) {
	o = parent.newFilterObservable("take")
	count = 0
	o.takeNum = num
	o.operator = takeOperator
	return o
}

// 只返回前面的N项数据，然后发射完成通知，忽略剩余的数据
var takeOperator = filterOperator{
	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
		count++
		if count > o.takeNum {
			return true
		}
		o.sendToFlow(ctx, item.Interface(), out)
		return false
	},
}

//TakeLast ： 获取最后n个元素
func (parent *Observable) TakeLast(num int) (o *Observable) {
	o = parent.newFilterObservable("takeLast")
	count = 0
	lasts = nil
	o.takeLastNum = num
	o.operator = takeLastOperator
	return o
}

// 只发射Observable'发射的后N项数据，忽略前面的数据
var takeLastOperator = filterOperator{
	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
		count++
		if count <= o.takeLastNum {
			lasts = append(lasts, item.Interface())
		} else {
			lasts = lasts[1:]
			lasts = append(lasts, item.Interface())
		}
		o.flip = lasts
		return false
	},
}
